//package bbkj.org.cn.demo;
//
//import java.io.Serializable;
//import java.math.BigDecimal;
//import java.util.List;
//import java.util.Map;
//import java.util.concurrent.atomic.AtomicLong;
//import java.util.concurrent.atomic.AtomicReference;
//
//import org.apache.flink.shaded.curator4.com.google.common.collect.Maps;
//import org.apache.flink.shaded.curator4.org.apache.curator.shaded.com.google.common.collect.Lists;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
//import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
//import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
//import org.apache.flink.util.Collector;
//
//import com.alibaba.fastjson.JSONObject;
//
//import bbkj.org.cn.sink.SinkCallback;
//import bbkj.org.cn.sink.WriteSink;
//import lombok.AllArgsConstructor;
//import lombok.Data;
//import lombok.extern.slf4j.Slf4j;
//
//@Slf4j
//public class WindowDemo implements Serializable{
//	
//	/**
//	 * 
//	 */
//	private static final long serialVersionUID = -3363260391295814964L;
//	public static String[] as = {"5","10","20","30"};
//	
//	public void convert(Item[] tmp,SinkCallback<Item> callback) {
//		LocalStreamEnvironment lse = new LocalStreamEnvironment();
//		DataStream<Item> flintstones = lse.fromElements(tmp);
//		
//		for(String a:as) {
//			flintstones = flintstones.countWindowAll(Integer.valueOf(a), 1).apply(new Avg(a));
//		}
//		
//		DataStream<Item> result = flintstones.countWindowAll(2, 1).apply(new LasterMap());
//		result.addSink(new WriteSink<Item>(callback));
//        try {
//			lse.execute();
//		} catch (Exception e) {
//			e.printStackTrace();
//		}
//	}
//	public static void main(String[] args) {
//		
//		Item[] tmp = JSONObject.parseArray("["
//				+ "{'c1':11.0},"
//				+ "{'c1':12.10},"
//				+ "{'c1':12.50},"
//				+ "{'c1':12.31},"
//				+ "{'c1':13.29},"
//				+ "{'c1':14.62},"
//				+ "{'c1':14.46},"
//				+ "{'c1':14.30},"
//				+ "{'c1':15.09},"
//				+ "{'c1':15.39},"
//				+ "{'c1':16.87},"
//				+ "{'c1':18.02},"
//				+ "{'c1':18.43},"
//				+ "{'c1':20.27},"
//				+ "{'c1':22.30},"
//				+ "{'c1':20.89},"
//				+ "{'c1':18.8},"
//				+ "{'c1':16.92},"
//				+ "{'c1':16.08},"
//				+ "{'c1':15.86},"
//				+ "{'c1':15.40},"
//				+ "{'c1':14.90},"
//				+ "{'c1':15.18}"
//				+ ",{'c1':14.5}"
//				+ ",{'c1':14.0}"
//				+ ",{'c1':13}"
//				+ "]",Item.class).toArray(new Item[0]);
//		
//		new WindowDemo().convert(tmp, new SinkCallback<WindowDemo.Item>() {
//			/**
//			 * 
//			 */
//			private static final long serialVersionUID = -8535832213343111038L;
//
//			@Override
//			public void callback(Item t) {
//				System.out.println(t);
//			}
//		});
//	}
//	@Data
//	public static class Item{
//		private BigDecimal c1;
//		private Map<String,BigDecimal> avg;
//		private Map<String,Object> status;
//	}
//	
//	@AllArgsConstructor
//	public static class LasterMap implements AllWindowFunction<WindowDemo.Item, WindowDemo.Item, GlobalWindow>{
//		/**
//		 * 
//		 */
//		private static final long serialVersionUID = 7224157738888136340L;
//
//		public void apply(GlobalWindow window, Iterable<Item> values, Collector<Item> out) throws Exception {
//			List<Item> list = Lists.newArrayList(values);
//			if(list.size()<2) {
//				out.collect(list.get(0));
//				return;
//			}
//			Item item = list.get(1);
//			Item itemPre = list.get(0);
//			if(item.getStatus() == null) {
//				item.setStatus(Maps.newHashMap());
//			}
//			if(itemPre!=null) {
//				boolean pass = item.getC1().compareTo(itemPre.getC1())>0;
//				item.getStatus().put("ups_downs", pass?"^":"v");
//				
//				for(String a:as) {
//					boolean pass5 = item.getC1().compareTo(itemPre.getAvg().get(a))>0;
//					item.getStatus().put("p_"+a, pass5?"gt":"lt");
//				}
//				for(String a:as) {
//					boolean pass20 = item.getAvg().get(a).compareTo(itemPre.getAvg().get(a))>0;
//					item.getStatus().put(a+"_ups_downs", pass20?"^":"v");
//				}
//				
//			}
//			out.collect(item);
//		}
//	}
//	
//	@AllArgsConstructor
//	public static class Avg implements AllWindowFunction<WindowDemo.Item, WindowDemo.Item, GlobalWindow>{
//
//		private String name;
//		/**
//		 * 
//		 */
//		private static final long serialVersionUID = 2012934163260954734L;
//
//		@Override
//		public void apply(GlobalWindow window, Iterable<Item> values, Collector<Item> out) throws Exception {
//			AtomicReference<BigDecimal> total = new AtomicReference<BigDecimal>(BigDecimal.ZERO);
//			AtomicLong al = new AtomicLong();
//			AtomicReference<Item> laster = new AtomicReference<Item>();
//			AtomicReference<Item> lasterPre = new AtomicReference<Item>();
//			values.forEach(item->{
//				total.set(total.get().add(item.getC1()));
//				al.incrementAndGet();
//				lasterPre.set(laster.get());
//				laster.set(item);
//			});
//			Item item = laster.get();
//			if(item.getAvg()==null) {
//				item.setAvg(Maps.newHashMap());
//			}
//			item.getAvg().put(name, total.get().divide(new BigDecimal(al.get()), 2, BigDecimal.ROUND_HALF_UP));
//			out.collect(item);
//		}
//	}
//}
